package org.jetlinks.pro.clickhouse.device;

import lombok.Generated;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.collections4.CollectionUtils;
import org.hswebframework.ezorm.core.param.TermType;
import org.hswebframework.ezorm.rdb.executor.SqlRequest;
import org.hswebframework.ezorm.rdb.executor.wrapper.ResultWrappers;
import org.hswebframework.ezorm.rdb.metadata.RDBDatabaseMetadata;
import org.hswebframework.ezorm.rdb.metadata.RDBTableMetadata;
import org.hswebframework.ezorm.rdb.operator.DatabaseOperator;
import org.hswebframework.ezorm.rdb.operator.DefaultDatabaseOperator;
import org.hswebframework.ezorm.rdb.operator.builder.fragments.SimpleTermsFragmentBuilder;
import org.hswebframework.ezorm.rdb.operator.ddl.TableBuilder;
import org.hswebframework.ezorm.rdb.operator.dml.query.Selects;
import org.hswebframework.web.api.crud.entity.PagerResult;
import org.hswebframework.web.api.crud.entity.QueryParamEntity;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.device.DeviceProductOperator;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.core.message.DeviceMessage;
import org.jetlinks.core.message.event.EventMessage;
import org.jetlinks.core.metadata.*;
import org.jetlinks.core.metadata.types.*;
import org.jetlinks.pro.ValueObject;
import org.jetlinks.pro.clickhouse.ClickHouseDataType;
import org.jetlinks.pro.clickhouse.ClickHouseOperations;
import org.jetlinks.pro.clickhouse.ClickHouseProperties;
import org.jetlinks.pro.clickhouse.engine.*;
import org.jetlinks.pro.clickhouse.metadata.ClickHouseDialect;
import org.jetlinks.pro.clickhouse.metadata.ClickHouseReactiveSqlExecutor;
import org.jetlinks.pro.clickhouse.metadata.ClickHouseSchemaMetadata;
import org.jetlinks.pro.clickhouse.types.ArrayDataType;
import org.jetlinks.pro.clickhouse.types.NestedDataType;
import org.jetlinks.pro.clickhouse.types.SimpleDataType;
import org.jetlinks.pro.device.entity.DeviceProperty;
import org.jetlinks.pro.device.service.data.AbstractDeviceDataStoragePolicy;
import org.jetlinks.pro.device.service.data.DeviceDataService;
import org.jetlinks.pro.device.service.data.DeviceDataStorageProperties;
import org.jetlinks.pro.device.timeseries.DeviceTimeSeriesMetric;
import org.jetlinks.pro.timeseries.TimeSeriesData;
import org.jetlinks.pro.timeseries.query.Aggregation;
import org.jetlinks.pro.timeseries.query.AggregationData;
import org.jetlinks.reactor.ql.utils.CastUtils;
import org.joda.time.DateTime;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
import org.springframework.util.Assert;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;

import javax.annotation.Nonnull;
import java.util.*;
import java.util.function.Consumer;
import java.util.function.Function;

/**
 * 使用ClickHouse来存储设备数据,一个属性为一行数据(行式存储)
 * <p>
 * 发布产品时,会自动创建表:properties_{产品ID}和device_log_{产品ID}.
 *
 * @author zhouhao
 * @since 1.9
 */
public class ClickHouseDeviceDataStoragePolicy extends AbstractDeviceDataStoragePolicy {

    private final ClickHouseOperations operations;

    private final DatabaseOperator databaseOperator;

    private final ClickHouseProperties clickHouseProperties;

    @Getter
    @Setter
    private PartitionInterval partitionInterval = PartitionInterval.days;

    public ClickHouseDeviceDataStoragePolicy(ClickHouseOperations operations,
                                             DeviceRegistry registry,
                                             DeviceDataStorageProperties properties,
                                             ClickHouseProperties clickHouseProperties) {
        super(registry, properties);
        this.operations = operations;
        this.clickHouseProperties = clickHouseProperties;

        RDBDatabaseMetadata database = new RDBDatabaseMetadata(new ClickHouseDialect());
        database.addFeature(new ClickHouseReactiveSqlExecutor(operations));

        ClickHouseSchemaMetadata schema = new ClickHouseSchemaMetadata(operations.getDatabase());

        database.addSchema(schema);
        database.setCurrentSchema(schema);
        databaseOperator = DefaultDatabaseOperator.of(database);
        setStrict(true);
    }


    @Override
    @Generated
    public String getId() {
        return "clickhouse";
    }

    @Override
    @Generated
    public String getName() {
        return "ClickHouse-行式存储";
    }

    @Override
    @Generated
    public String getDescription() {
        return "使用ClickHouse存储设备数据,每一个属性为一行数据.";
    }

    @Nonnull
    @Override
    @Generated
    public Mono<ConfigMetadata> getConfigMetadata() {
        return Mono.empty();
    }

    @Override
    protected Flux<Tuple2<String, TimeSeriesData>> convertProperties(String productId,
                                                                     DeviceMessage message,
                                                                     Map<String, Object> properties,
                                                                     Map<String, Long> propertySourceTimes) {
        return convertPropertiesForRowPolicy(productId, message, properties, propertySourceTimes);
    }

    @Override
    protected Map<String, Object> createRowPropertyData(String id,
                                                        long timestamp,
                                                        String deviceId,
                                                        PropertyMetadata property,
                                                        Object value) {
        Map<String, Object> values = super.createRowPropertyData(id, timestamp, deviceId, property, value);

        //时间统一转换为时间戳
        values.compute("timeValue", (key, val) -> {
            if (val == null) {
                return null;
            }
            return CastUtils.castDate(val).getTime();
        });
        //地理位置值存储为数组
        values.compute("geoValue", (key, val) -> {
            if (val == null) {
                return null;
            }
            GeoPoint point = GeoPoint.of(val);

            return Arrays.asList((float)point.getLon(), (float)point.getLat());
        });

        //默认null
        values.putIfAbsent("numberValue", null);
        values.putIfAbsent("timeValue", null);
        values.putIfAbsent("geoValue", null);
        //以下属性不保存
        values.remove("type");
        values.remove("objectValue");
        values.remove("arrayValue");

        return values;
    }

    @Override
    protected Map<String, Object> createEventData(EventMessage message, DeviceMetadata metadata) {

        return convertValue(super.createEventData(message, metadata));
    }

    @Override
    protected Map<String, Object> createDeviceLogData(String productId, DeviceMessage message) {
        Map<String, Object> data = super.createDeviceLogData(productId, message);
        data.remove("productId");
        return convertValue(data);
    }

    private Map<String, Object> convertValue(Map<String, Object> value) {
        for (Map.Entry<String, Object> stringObjectEntry : value.entrySet()) {
            if (stringObjectEntry.getValue() instanceof Date) {
                stringObjectEntry.setValue(((Date) stringObjectEntry.getValue()).getTime());
            }
            if (stringObjectEntry.getValue() instanceof GeoPoint) {
                GeoPoint point = ((GeoPoint) stringObjectEntry.getValue());
                stringObjectEntry.setValue(Arrays.asList((float)point.getLon(),(float)point.getLat()));
            }
        }
        return value;
    }

    @Override
    protected <T> Flux<T> doQuery(String metric,
                                  QueryParamEntity paramEntity,
                                  Function<TimeSeriesData, T> mapper) {
        if (paramEntity.getPageSize() == 0) {
            return Flux.empty();
        }
        return databaseOperator
            .dml()
            .query(createQueryTable(metric))
            .setParam(paramEntity)
            .fetch(ResultWrappers.map())
            .reactive()
            .map(map -> {
                long ts = CastUtils.castNumber(map.get("timestamp")).longValue();
                return mapper.apply(TimeSeriesData.of(ts, map));
            });
    }

    @Override
    protected <T> Mono<PagerResult<T>> doQueryPager(String metric,
                                                    QueryParamEntity paramEntity,
                                                    Function<TimeSeriesData, T> mapper) {
        return Mono
            .zip(
                //count
                Mono
                    .justOrEmpty(paramEntity.getTotal())
                    .switchIfEmpty(databaseOperator
                                       .dml()
                                       .query(createQueryTable(metric))
                                       .select(Selects.count1().as("total"))
                                       .setParam(paramEntity.clone().noPaging().doNotSort())
                                       .fetch(ResultWrappers.map())
                                       .reactive()
                                       .map(map -> CastUtils.castNumber(map.getOrDefault("total", 0)).intValue())
                                       .singleOrEmpty())
                    .defaultIfEmpty(0),
                //select
                this
                    .doQuery(metric, paramEntity, mapper)
                    .collectList(),
                (total, data) -> PagerResult.of(total, data, paramEntity)
            );
    }

    private Engine createEngine(Consumer<MergeTree> mergeTreeConsumer) {
        return clickHouseProperties
            .getTemplate()
            .builder()
            .when(MergeTree.class, mergeTreeConsumer)
            .build();
    }


    @Override
    protected Mono<Void> doSaveData(String metric, TimeSeriesData data) {
        return operations.insert(metric, data.values());
    }

    @Override
    protected Mono<Void> doSaveData(String metric, Flux<TimeSeriesData> data) {
        return operations.insert(metric, data.map(TimeSeriesData::values));
    }

    private Mono<Void> createLogTable(String tableName, Engine engine) {
        return databaseOperator
            .ddl()
            .createOrAlter(tableName)
            .addColumn("id").type(SimpleDataType.String).comment("ID").commit()
            .addColumn("timestamp").type(SimpleDataType.Int64).comment("上报时间").commit()
            .addColumn("createTime").type(SimpleDataType.Int64).comment("创建时间").commit()
            .addColumn("content").type(SimpleDataType.String).comment("日志内容").commit()
            .addColumn("deviceId").type(SimpleDataType.String).comment("设备ID").commit()
            .addColumn("messageId").type(SimpleDataType.String).comment("消息ID").commit()
            .addColumn("type").type(SimpleDataType.String).comment("日志类型").commit()
            .custom(table -> {
                //配置表引擎
                table.addFeature(engine);
            })
            .commit()
            .reactive()
            .then();
    }

    private Mono<Void> createPropertiesTable(String tableName, Engine engine) {
        return databaseOperator
            .ddl()
            .createOrAlter(tableName)
            .addColumn("id").type(SimpleDataType.String).comment("ID").commit()
            .addColumn("timestamp").type(SimpleDataType.Int64).comment("上报时间").commit()
            .addColumn("createTime").type(SimpleDataType.Int64).comment("创建时间").commit()
            .addColumn("timeValue").type(SimpleDataType.Int64).comment("时间类型值").commit()
            .addColumn("property").type(SimpleDataType.String).comment("属性ID").commit()
            .addColumn("deviceId").type(SimpleDataType.String).comment("设备ID").commit()
            .addColumn("value").type(SimpleDataType.String).comment("值").commit()
            .addColumn("numberValue").type(SimpleDataType.Float64).comment("数字类型值").commit()
            .addColumn("geoValue").type(ArrayDataType.of(SimpleDataType.Float32)).comment("地理位置值").commit()
            .custom(table -> {
                //配置表引擎
                table.addFeature(engine);
            })
            .commit()
            .reactive()
            .then();
    }

    private Mono<Void> createEventTable(String tableName, EventMetadata event, Engine engine) {

        TableBuilder tableBuilder = databaseOperator
            .ddl()
            .createOrAlter(tableName)
            .addColumn("id").type(SimpleDataType.String).comment("ID").commit()
            .addColumn("deviceId").type(SimpleDataType.String).comment("设备ID").commit()
            .addColumn("timestamp").type(SimpleDataType.Int64).comment("上报时间").commit()
            .addColumn("createTime").type(SimpleDataType.Int64).comment("创建时间").commit()
            .custom(table -> {
                //配置表引擎
                table.addFeature(engine);
            });

        DataType type = event.getType();
        if (type instanceof ObjectType) {
            //对象类型则平铺
            for (PropertyMetadata property : ((ObjectType) type).getProperties()) {
                tableBuilder.addColumn(property.getId())
                            .type(convertCHType(property.getValueType()))
                            .comment(property.getName())
                            .commit();
            }
        } else {
            tableBuilder.addColumn("value")
                        .type(convertCHType(type))
                        .comment("事件值")
                        .commit();
        }

        return tableBuilder
            .commit()
            .reactive()
            .then();
    }

    private static final Map<String, ClickHouseDataType> typeMapping = new HashMap<>();

    static {
        typeMapping.put(IntType.ID, SimpleDataType.Int32);
        typeMapping.put(LongType.ID, SimpleDataType.Int64);
        typeMapping.put(FloatType.ID, SimpleDataType.Float32);
        typeMapping.put(DoubleType.ID, SimpleDataType.Float64);
        typeMapping.put(BooleanType.ID, SimpleDataType.Int8);

        typeMapping.put(DateTimeType.ID, SimpleDataType.Int64);

        typeMapping.put(StringType.ID, SimpleDataType.String);
        typeMapping.put(PasswordType.ID, SimpleDataType.String);
        typeMapping.put(EnumType.ID, SimpleDataType.String);
        typeMapping.put(FileType.ID, SimpleDataType.String);

        typeMapping.put(GeoType.ID, ArrayDataType.of(SimpleDataType.Float32));

    }

    ClickHouseDataType convertCHType(DataType dataType) {
        ClickHouseDataType mapped = typeMapping.get(dataType.getType());
        if (null != mapped) {
            return mapped;
        }
        if (dataType instanceof ArrayType) {
            DataType elementType = ((ArrayType) dataType).getElementType();
            return ArrayDataType.of(convertCHType(elementType));
        }

        if (dataType instanceof ObjectType) {
            List<PropertyMetadata> elementType = ((ObjectType) dataType).getProperties();
            NestedDataType nestedDataType = new NestedDataType();
            for (PropertyMetadata propertyMetadata : elementType) {
                nestedDataType.field(propertyMetadata.getId(), convertCHType(propertyMetadata.getValueType()));
            }
            return nestedDataType;
        }

        throw new UnsupportedOperationException("不支持的类型:" + dataType.getType());
    }

    protected boolean isDistributedQuery() {
        Engine engine = clickHouseProperties.getTemplate().currentEngine();

        return !(engine instanceof Distributed) && engine.isDistributed();
    }

    protected String createDistributedQueryTable(String nativeTable) {
        return nativeTable + "_all";
    }

    protected String createQueryTable(String nativeTable) {
        if (isDistributedQuery()) {
            return createDistributedQueryTable(nativeTable);
        }
        return nativeTable;
    }

    @Override
    protected String getDeviceEventMetric(String productId, String eventId) {
        return super.getDeviceEventMetric(productId, eventId).replace("-", "_");
    }

    @Nonnull
    @Override
    public Mono<Void> registerMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {

        String propertiesTable = getPropertyTimeSeriesMetric(productId);
        String logTable = getDeviceLogMetric(productId);

        List<Mono<Void>> jobs = new ArrayList<>(6);

        //日志表
        {
            Engine logEngine = createEngine(mergeTree -> mergeTree
                .orderBy("(deviceId,timestamp)")
                .partitionBy(partitionInterval.create("timestamp/1000")));


            jobs.add(createLogTable(logTable, logEngine));
            if (isDistributedQuery()) {
                jobs.add(createLogTable(createDistributedQueryTable(logTable),
                                        new Distributed()
                                            .table(logTable)
                                            .cluster(logEngine.getCluster())
                                            .database(operations.getDatabase())));
            }
        }

        //属性表
        {
            Engine propertiesEngine = createEngine(mergeTree -> mergeTree
                .orderBy("(deviceId,property,timestamp)")
//                .primaryKey("(deviceId,property)")
                .partitionBy(partitionInterval.create("timestamp/1000")));
            jobs.add(createPropertiesTable(propertiesTable, propertiesEngine));
            if (isDistributedQuery()) {
                jobs.add(createPropertiesTable(createDistributedQueryTable(propertiesTable),
                                               new Distributed()
                                                   .table(propertiesTable)
                                                   .cluster(propertiesEngine.getCluster())
                                                   .database(operations.getDatabase())));
            }
        }
        //事件表,一个事件一个表
        for (EventMetadata event : metadata.getEvents()) {
            String eventTable = getDeviceEventMetric(productId, event.getId());
            Engine engine = createEngine(mergeTree -> mergeTree
                .orderBy("(deviceId)")
                .partitionBy(partitionInterval.create("timestamp/1000")));
            jobs.add(createEventTable(eventTable, event, engine));
            if (isDistributedQuery()) {
                jobs.add(createEventTable(createDistributedQueryTable(eventTable), event,
                                          new Distributed()
                                              .table(eventTable)
                                              .cluster(engine.getCluster())
                                              .database(operations.getDatabase())));
            }
        }

        return Flux
            .concat(jobs)
            .then(reloadMetadata(productId, metadata));
    }

    @Override
    protected String getDeviceLogMetric(String productId) {
        return super.getDeviceLogMetric(productId.replace("-", "_").replace(".", "_").toLowerCase());
    }

    @Override
    protected String getPropertyTimeSeriesMetric(String productId) {
        return super.getPropertyTimeSeriesMetric(productId.replace("-", "_").replace(".", "_").toLowerCase());
    }

    @Nonnull
    @Override
    public Mono<Void> reloadMetadata(@Nonnull String productId, @Nonnull DeviceMetadata metadata) {
        List<String> tables = new ArrayList<>();
        tables.add(getPropertyTimeSeriesMetric(productId));
        tables.add(getDeviceLogMetric(productId));

        if (isDistributedQuery()) {
            tables.add(createQueryTable(getPropertyTimeSeriesMetric(productId)));
            tables.add(createQueryTable(getDeviceLogMetric(productId)));
        }

        return Flux
            .fromIterable(tables)
            .flatMap(databaseOperator
                         .getMetadata()
                         .getCurrentSchema()::getTableReactive)
            .then();
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryEachOneProperties(@Nonnull String deviceId,
                                                       @Nonnull QueryParamEntity query,
                                                       @Nonnull String... properties) {

        return this
            .getProductAndMetadataByDevice(deviceId)
            .flatMapMany(tp2 -> {
                DeviceMetadata metadata = tp2.getT2();
                Set<String> props = new HashSet<>(Arrays.asList(properties));
                return Flux
                    .fromIterable(metadata.getProperties())
                    .filter(prop -> props.isEmpty() || props.contains(prop.getId()))
                    .flatMap(prop -> this
                        .queryProperty(tp2,
                                       query
                                           .clone()
                                           .doPaging(0, 1)
                                           .and("deviceId", TermType.eq, deviceId),
                                       prop.getId()));
            });
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryEachProperties(@Nonnull String deviceId,
                                                    @Nonnull QueryParamEntity query,
                                                    @Nonnull String... property) {
        Set<String> properties = new HashSet<>(Arrays.asList(property));
        return this
            .getProductAndMetadataByDevice(deviceId)
            .flatMapMany(tp2 -> {
                DeviceMetadata metadata = tp2.getT2();
                return Flux
                    .fromIterable(metadata.getProperties())
                    .filter(prop -> properties.isEmpty() || properties.contains(prop.getId()))
                    .map(prop -> this.queryProperty(tp2,
                                                    query.clone().and("deviceId", TermType.eq, deviceId),
                                                    prop.getId()))
                    .buffer(20)
                    .concatMap(Flux::merge);
            });
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryProperty(@Nonnull String deviceId,
                                              @Nonnull QueryParamEntity query,
                                              @Nonnull String... properties) {
        return this
            .getProductAndMetadataByDevice(deviceId)
            .flatMapMany(tp2 -> queryProperty(tp2, query.clone().and("deviceId", TermType.eq, deviceId), properties));

    }

    private Flux<DeviceProperty> queryProperty(Tuple2<DeviceProductOperator, DeviceMetadata> tp2,
                                               QueryParamEntity query,
                                               String... properties) {
        String table = createQueryTable(getPropertyTimeSeriesMetric(tp2.getT1().getId()));

        DeviceMetadata metadata = tp2.getT2();
        return query
            .clone()
            .toQuery()
            //指定了属性，则按指定的属性数量进行分页
            .when(query.isPaging() && properties.length > 0, q -> q
                .doPaging(query.getPageIndex(), properties.length * query.getPageSize())
                .in("property", (Object[]) properties))
            //没有指定属性，则按全部属性数量进行分页
            .when(query.isPaging() && properties.length == 0, q -> q
                .doPaging(query.getPageIndex(), metadata.getProperties().size() * query.getPageSize()))
            .when(CollectionUtils.isEmpty(query.getSorts()), q -> q.orderByDesc("timestamp"))
            .execute(databaseOperator.dml().query(table)::setParam)
            .fetch(ResultWrappers.map())
            .reactive()
            .map(map -> {
                String property = String.valueOf(map.get("property"));
                long timestamp = CastUtils.castNumber(map.get("timestamp")).longValue();

                return DeviceProperty
                    .of(TimeSeriesData.of(timestamp, map), metadata.getPropertyOrNull(property));
            });
    }

    @Nonnull
    @Override
    public Flux<DeviceProperty> queryPropertyByProductId(@Nonnull String productId,
                                                         @Nonnull QueryParamEntity query,
                                                         @Nonnull String... properties) {
        return this
            .getProductAndMetadataByProduct(productId)
            .flatMapMany(tp2 -> queryProperty(tp2, query, properties));
    }


    protected String createAggFunction(Aggregation aggregation) {
        switch (aggregation) {
            case COUNT:
                return "count(id)";
            case AVG:
                return "avg(numberValue)";
            case MAX:
                return "max(numberValue)";
            case MIN:
                return "min(numberValue)";
            case SUM:
                return "sum(numberValue)";
            case FIRST:
                return "first_value(numberValue)";
            case MEDIAN:
                return "median(numberValue)";

        }
        throw new UnsupportedOperationException("不支持的聚合函数:" + aggregation);
    }

    @Override
    public Flux<AggregationData> aggregationPropertiesByProduct(@Nonnull String productId,
                                                                @Nonnull DeviceDataService.AggregationRequest request,
                                                                @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
        Map<Long, Map<String, Object>> dataReference = aggCompleteResultReference(request, properties);
        String table = createQueryTable(getPropertyTimeSeriesMetric(productId));
        boolean groupByTime = request.getInterval() != null;

        StringBuilder sql = new StringBuilder();
        if (request.getInterval() != null) {
            long interval = request.getInterval().toMillis();
            String SQL = "select round(timestamp/%d)*%d time,property";
            sql.append(String.format(SQL, interval, interval));
        } else {
            sql.append("select property");
        }
        Set<String> propertyId = new HashSet<>();

        for (DeviceDataService.DevicePropertyAggregation property : properties) {
            sql.append(",if(property='").append(property.getProperty()).append("',");
            sql
                .append(createAggFunction(property.getAgg()))
                .append(",null) \"")
                .append(property.getAlias())
                .append("\"");
            propertyId.add(property.getProperty());
        }
        if (!propertyId.isEmpty()) {
            request.getFilter().and("property", TermType.in, propertyId);
        }
        request.getFilter().and("timestamp", TermType.btw, Arrays.asList(request.getFrom(), request.getTo()));

        sql.append(" from ").append(table);

        return databaseOperator
            .getMetadata()
            .getTableOrViewReactive(table)
            .cast(RDBTableMetadata.class)
            .flatMapMany(tableMeta -> {
                SqlRequest sqlRequest = SimpleTermsFragmentBuilder
                    .instance()
                    .createTermFragments(tableMeta, request.getFilter().getTerms())
                    .toRequest();
                if (sqlRequest.isNotEmpty()) {
                    sql.append(" where ").append(sqlRequest.toNativeSql());
                }
                sql.append(" group by property");
                if (groupByTime) {
                    sql.append(",time");
                }
                return operations
                    .query(sql.toString(), ResultWrappers.map())
                    .map(ValueObject::of);
            })
            //按时间分组
            .groupBy(value -> value.getLong("time").orElse(0L), Integer.MAX_VALUE)
            .flatMap(group -> group
                .doOnNext(value -> {
                    for (DeviceDataService.DevicePropertyAggregation property : properties) {
                        value.get(property.getAlias())
                             .ifPresent(val -> {
                                 Optional.ofNullable(dataReference.get(group.key()))
                                         .ifPresent(data -> data.put(property.getAlias(), val));
                             });
                    }
                })
            )
            .thenMany(Flux.fromIterable(dataReference.values()))
            .map(AggregationData::of);
    }


    /**
     * 获取完整数据参考
     */
    private Map<Long, Map<String, Object>> aggCompleteResultReference(DeviceDataService.AggregationRequest request,
                                                                      DeviceDataService.DevicePropertyAggregation... properties) {
        DateTimeFormatter formatter = DateTimeFormat.forPattern(request.getFormat());
        Assert.notNull(request.getInterval(), "聚合时间间隔不能为空");
        long interval = request.getInterval().toMillis();
        Map<Long, Map<String, Object>> data = new LinkedHashMap<>();
        Map<String, Object> valueMap = new HashMap<>();
        long endTimestamp = (request.getTo().getTime() / interval) * interval;
        for (DeviceDataService.DevicePropertyAggregation property : properties) {
            valueMap.put(property.getAlias(), 0);
        }
        // TODO: 2021/5/20 加入排序后，处理逻辑将变更
        //默认倒叙排列
        for (long i = endTimestamp; i > request.getFrom().getTime(); i -= interval) {
            String formatValue = new DateTime(new Date(i)).toString(formatter);
            Map<String, Object> tempData = new HashMap<>(valueMap);
            tempData.put("time", formatValue);
            data.put(i, tempData);
            if (request.getLimit() == data.size()) {
                return data;
            }
        }
        return data;
    }

    @Override
    public Flux<AggregationData> aggregationPropertiesByDevice(@Nonnull String deviceId,
                                                               @Nonnull DeviceDataService.AggregationRequest request,
                                                               @Nonnull DeviceDataService.DevicePropertyAggregation... properties) {
        return deviceRegistry
            .getDevice(deviceId)
            .flatMap(DeviceOperator::getProduct)
            .flatMapMany(product -> {
                DeviceDataService.AggregationRequest newRequest = request.copy();
                newRequest.getFilter().and("deviceId", TermType.eq, deviceId);
                return aggregationPropertiesByProduct(product.getId(), newRequest, properties);
            });
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
                                                               @Nonnull String property,
                                                               @Nonnull QueryParamEntity param) {
        return getProductAndMetadataByDevice(deviceId)
            .flatMap(productAndMetadata -> queryPropertyPage(productAndMetadata, param.and("deviceId", TermType.eq, deviceId), property));
    }


    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPageByProductId(@Nonnull String productId,
                                                                          @Nonnull String property,
                                                                          @Nonnull QueryParamEntity query) {
        return queryPropertyPageByProductId(productId,query,property);
    }


    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull String deviceId,
                                                               @Nonnull QueryParamEntity query,
                                                               @Nonnull String... properties) {
        return getProductAndMetadataByDevice(deviceId)
            .flatMap(productAndMetadata -> queryPropertyPage(productAndMetadata, query.and("deviceId", TermType.eq, deviceId), properties));
    }

    @Nonnull
    @Override
    public Mono<PagerResult<DeviceProperty>> queryPropertyPageByProductId(@Nonnull String productId,
                                                                          @Nonnull QueryParamEntity query,
                                                                          @Nonnull String... properties) {
        return getProductAndMetadataByProduct(productId)
            .flatMap(productAndMetadata -> queryPropertyPage(productAndMetadata, query, properties));
    }

    private Mono<PagerResult<DeviceProperty>> queryPropertyPage(@Nonnull Tuple2<DeviceProductOperator, DeviceMetadata> productAndMetadata,
                                                                @Nonnull QueryParamEntity query,
                                                                @Nonnull String... property) {

        String table = getPropertyTimeSeriesMetric(productAndMetadata.getT1().getId());

        return query
            .toQuery()
            .when(property.length > 0, q -> q.in("property", Arrays.asList(property)))
            .execute(param -> this
                .doQueryPager(table, query, data -> DeviceProperty.of(data, data
                    .getString("property")
                    .map(productAndMetadata.getT2()::getPropertyOrNull)
                    .orElse(null)
                )));
    }

    @Override
    protected DeviceDataStorageProperties.Event eventSetting() {
        // FIXME: 2021/12/6 支持相关事件的自定义逻辑
        return DeviceDataStorageProperties.Event.DEFAULT;
    }
}
